package com.stable.service;

import com.alibaba.fastjson.JSON;
import com.google.common.util.concurrent.ListenableFuture;
import com.stable.enums.RunCycleEnum;
import com.stable.enums.RunLogBizTypeEnum;
import com.stable.es.dao.base.EsProBarInfoDao;
import com.stable.job.MyCallable;
import com.stable.spider.tushare.TushareSpider;
import com.stable.utils.RedisUtil;
import com.stable.utils.TasksWorker;
import com.stable.vo.bus.HorizontalInfo;
import com.stable.vo.bus.ProBarInfo;
import com.stable.vo.bus.StockBaseInfo;
import com.stable.vo.spi.req.EsQueryPageReq;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service("HorizontalService")
@Log4j2
public class HorizontalService {
    @Autowired
    private EsProBarInfoDao esProBarInfoDao;

    @Autowired
    private TushareSpider tushareSpider;

    @Autowired
    private RedisUtil redisUtil;

    private final Semaphore semapProBar = new Semaphore(1);

    // 把水平涨跌弧度数据写入redis
    public ListenableFuture<Object> jobQueryHorizontalToRedis(String queryDate, String startDate, String endDate, String code, EsQueryPageReq queryPage,
                                                              int startStepDays,
                                                              int startStepFullLimitCount,
                                                              int startStepRange,
                                                              int calculateAvgCountN,
                                                              int avgNDownRange,
                                                              int avgNDownRMoreExcludeCount,
                                                              int avgNUpRange,
                                                              int avgNUpRMoreExcludeCount,
                                                              int moreThanMa250Range,
                                                              int near7daysRange,
                                                              int topToNDownRange) {

        if (queryPage.getPageSize() == 0) queryPage.setPageSize(1000); //默认取1只股票3年数据
        log.info("page startDate={},endDate={},pageable.page={},pageable.size={}, sortName:{} ,sortOrder:{}", startDate, endDate, queryPage.getPageNum(),
                queryPage.getPageSize(), queryPage.getSortName(), queryPage.getSortOrder());
        return TasksWorker.getInstance().getService()
                .submit(new MyCallable(RunLogBizTypeEnum.PRO_BAR_HORIZONTAL, RunCycleEnum.DAY) {
                    public Object mycall() {
                        return doHorizontalMethod(queryDate, startDate, endDate, code, queryPage, startStepDays, startStepFullLimitCount, startStepRange, calculateAvgCountN, avgNDownRange, avgNDownRMoreExcludeCount, avgNUpRange, avgNUpRMoreExcludeCount, moreThanMa250Range, near7daysRange, topToNDownRange);
                    }
                });
    }

    public Object doHorizontalMethod(String queryDate, String startDate, String endDate, String code, EsQueryPageReq queryPage, int startStepDays, int startStepFullLimitCount, int startStepRange, int calculateAvgCountN, int avgNDownRange, int avgNDownRMoreExcludeCount, int avgNUpRange, int avgNUpRMoreExcludeCount, int moreThanMa250Range, int near7daysRange, int topToNDownRange) {
        try {
            semapProBar.acquire();
            String redisMapKey = "horizontal_" + queryDate + "_" + startDate + "_" + endDate + "_" + near7daysRange;

            Map<Object, Object> stockBasicMap = redisUtil.hGetAll("stock_basic");
            if (StringUtils.isNotBlank(code)) {
                StockBaseInfo stockBaseInfo = JSON.parseObject((String) stockBasicMap.get(code), StockBaseInfo.class);
                if (!Objects.isNull(stockBaseInfo)) {
                    String baseCode = stockBaseInfo.getCode();
                    String conceptName = stockBaseInfo.getConcept_name();
                    doCodeAnalyMethod(queryDate, startDate, endDate, code, queryPage, startStepDays, startStepFullLimitCount, startStepRange, calculateAvgCountN, avgNDownRange, avgNDownRMoreExcludeCount, avgNUpRange, avgNUpRMoreExcludeCount, moreThanMa250Range, near7daysRange, redisMapKey, baseCode, stockBaseInfo, topToNDownRange);
                }
            } else {
                stockBasicMap.values().stream().forEach(info -> {
                    StockBaseInfo stockBaseInfo = JSON.parseObject((String) info, StockBaseInfo.class);
                    if (!Objects.isNull(stockBaseInfo)) {
                        String baseCode = stockBaseInfo.getCode();
                        String conceptName = stockBaseInfo.getConcept_name();
                        doCodeAnalyMethod(queryDate, startDate, endDate, code, queryPage, startStepDays, startStepFullLimitCount, startStepRange, calculateAvgCountN, avgNDownRange, avgNDownRMoreExcludeCount, avgNUpRange, avgNUpRMoreExcludeCount, moreThanMa250Range, near7daysRange, redisMapKey, baseCode, stockBaseInfo, topToNDownRange);
                    }
                });
            }
            redisUtil.expire(redisMapKey, 7, TimeUnit.DAYS);
            return null;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        } finally {
            semapProBar.release();
        }
    }

    private boolean doCodeAnalyMethod(String queryDate, String startDate, String endDate, String code, EsQueryPageReq queryPage, int startStepDays, int startStepFullLimitCount, int startStepRange, int calculateAvgCountN, int avgNDownRange, int avgNDownRMoreExcludeCount, int avgNUpRange, int avgNUpRMoreExcludeCount, int moreThanMa250Range, int near7daysRange, String redisMapKey, String baseCode, StockBaseInfo baseInfo, int topToNDownRange) {
        Pageable pageable = PageRequest.of(queryPage.getPageNum(), queryPage.getPageSize());
        BoolQueryBuilder bqb = QueryBuilders.boolQuery();
        if (StringUtils.isNotBlank(baseCode)) {
            bqb.must(QueryBuilders.matchPhraseQuery("code", baseCode));
        }
        bqb.filter(QueryBuilders.rangeQuery("tradeDate").gte(startDate).lte(endDate));
        FieldSortBuilder sort = SortBuilders.fieldSort(StringUtils.isEmpty(queryPage.getSortName()) ? "tradeDate" : queryPage.getSortName()).unmappedType("integer").order(SortOrder.DESC);

        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        SearchQuery sq = queryBuilder.withQuery(bqb).withSort(sort).withPageable(pageable).build();
        List<ProBarInfo> proBarInfos = esProBarInfoDao.search(sq).getContent();

        if (proBarInfos.size() <= startStepDays) return true;
        //开始时间起的4个月（85个交易日）最高最低价格
        int start85 = startStepDays;
        double maxStart85Close = proBarInfos.stream().skip(proBarInfos.size() - start85).collect(Collectors.maxBy(Comparator.comparingDouble(ProBarInfo::getClosed))).get().getClosed();
        double minStart85Close = proBarInfos.stream().skip(proBarInfos.size() - start85).collect(Collectors.minBy(Comparator.comparingDouble(ProBarInfo::getClosed))).get().getClosed();
        if (((float) (maxStart85Close - minStart85Close) / minStart85Close) * 100 < startStepRange)
            return true;

        AtomicInteger fullLimitUpCount = new AtomicInteger(); //记录涨停次数,如果没有涨停的直接抛弃
        proBarInfos.stream().skip(proBarInfos.size() - start85).forEach(info -> {
            if (info.getChangeRate() > 9.9) {
                fullLimitUpCount.incrementAndGet();
            }
        });
        if (fullLimitUpCount.get() < startStepFullLimitCount) return true;


        //把每10个交易日分为一组；检查各组数据平均价格间隔
        int team = (proBarInfos.size() - start85) / calculateAvgCountN;
        double avg10Close[] = new double[team];
        for (int b = 0; b < team; b++) {
            avg10Close[b] = proBarInfos.stream().skip((b) * calculateAvgCountN).limit(calculateAvgCountN).collect(Collectors.averagingDouble(ProBarInfo::getClosed));
        }

        // 剔除最迟的85个交易日，计算剩下的所有交易日的平均价格
        Double avgExclude85DaysClose = proBarInfos.stream().limit(proBarInfos.size() - start85).collect(Collectors.averagingDouble(ProBarInfo::getClosed));

        // 记录超过涨幅、跌幅范围 数量
        int upAvgRangeCount = 0, downAvgRangeCount = 0;
        for (int c = 0; c < team; c++) {
            double avgClose = avg10Close[c];
            // 10日平均价 与 剔除最迟85日的平均价格，相差的百分比
            double v = ((float) (avgClose - avgExclude85DaysClose) / avgExclude85DaysClose) * 100;

            log.debug("code:{} avgAllClose:{}: 10avg: {}, v:{}", baseCode, avgExclude85DaysClose, avgClose, v);
            if (v < avgNDownRange) {
                downAvgRangeCount++;
            }
            if (v > avgNUpRange) {
                upAvgRangeCount++;
            }
        }
        log.info("code：{} 高于百分比：{}的量为：upAvgRangeCount: {},低于百分比：{}的量为：downAvgRangeCount:{}", code, avgNUpRange, upAvgRangeCount, avgNDownRange, downAvgRangeCount);
        if (downAvgRangeCount > avgNDownRMoreExcludeCount) return true;
        if (upAvgRangeCount > avgNUpRMoreExcludeCount) return true;


        AtomicInteger moreThanMa250Count = new AtomicInteger(); //250日线上的量
        AtomicInteger lessThanMa250Count = new AtomicInteger(); //250日线下的量
        proBarInfos.stream().limit(proBarInfos.size() - start85).forEach(info -> {
            if (info.getClosed() > info.getMa250()) {
                moreThanMa250Count.incrementAndGet();
            } else {
                log.debug("data:{}, close:{}, ma250:{}", info.getTradeDate(), info.getClosed(), info.getMa250());
                lessThanMa250Count.incrementAndGet();
            }
        });
        log.info("code:{},250日线上的量:{},250日线下的量:{}", baseCode, moreThanMa250Count, lessThanMa250Count);
        if (((float) moreThanMa250Count.get() / (proBarInfos.size() - start85)) * 100 < moreThanMa250Range)
            return true;

        //除开开始阶段，最低价
        double minStart85AfterClose = proBarInfos.stream().limit(proBarInfos.size() - start85).collect(Collectors.minBy(Comparator.comparingDouble(ProBarInfo::getClosed))).get().getClosed();
        //计算开始阶段最高价 与 水平阶段最低价，不能下跌超过35%
        if (((float) (maxStart85Close - minStart85AfterClose) / maxStart85Close) * 100 > topToNDownRange)
            return true;

        //获取最近5日的收盘价
        double pre1Close = proBarInfos.get(0).getClosed();
        double pre2Close = proBarInfos.get(1).getClosed();
        double pre3Close = proBarInfos.get(2).getClosed();
        double pre4Close = proBarInfos.get(3).getClosed();
        double pre5Close = proBarInfos.get(4).getClosed();
        double pre6Close = proBarInfos.get(5).getClosed();
        double pre7Close = proBarInfos.get(6).getClosed();
        log.info("最近5日收盘价 1:{}，2:{}，3:{}，4:{}，5:{}，6:{}，7:{}", pre1Close, pre2Close, pre3Close, pre4Close, pre5Close, pre6Close, pre7Close);

        AtomicInteger pre1ConformCount = new AtomicInteger(); //第1天相差少于指定百分比的量
        AtomicInteger pre2ConformCount = new AtomicInteger(); //第2天相差少于指定百分比的量
        AtomicInteger pre3ConformCount = new AtomicInteger(); //第3天相差少于指定百分比的量
        AtomicInteger pre4ConformCount = new AtomicInteger(); //第4天相差少于指定百分比的量
        AtomicInteger pre5ConformCount = new AtomicInteger(); //第5天相差少于指定百分比的量
        AtomicInteger pre6ConformCount = new AtomicInteger(); //第6天相差少于指定百分比的量
        AtomicInteger pre7ConformCount = new AtomicInteger(); //第7天相差少于指定百分比的量

        AtomicInteger incrementCount = new AtomicInteger(); //在统计范围内，上涨的天数
        proBarInfos.stream().forEach(info -> {
            double closed = info.getClosed();
            int tradeDate = info.getTradeDate();
            double v1 = ((pre1Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第1天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre1Close, closed, v1);
            if (Math.abs(v1) <= near7daysRange) pre1ConformCount.incrementAndGet();

            double v2 = ((pre2Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第2天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre2Close, closed, v2);
            if (Math.abs(v2) <= near7daysRange) pre2ConformCount.incrementAndGet();

            double v3 = ((pre3Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第3天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre3Close, closed, v3);
            if (Math.abs(v3) <= near7daysRange) pre3ConformCount.incrementAndGet();

            double v4 = ((pre4Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第4天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre4Close, closed, v4);
            if (Math.abs(v4) <= near7daysRange) pre4ConformCount.incrementAndGet();

            double v5 = ((pre5Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第5天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre5Close, closed, v5);
            if (Math.abs(v5) <= near7daysRange) pre5ConformCount.incrementAndGet();

            double v6 = ((pre6Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第6天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre6Close, closed, v6);
            if (Math.abs(v6) <= near7daysRange) pre6ConformCount.incrementAndGet();

            double v7 = ((pre7Close - closed) / closed) * 100;
            log.debug("交易日期：{}, 第7天价格：{}，当前价格：{}, 相差百分比:{}", tradeDate, pre7Close, closed, v7);
            if (Math.abs(v7) <= near7daysRange) pre7ConformCount.incrementAndGet();


            // 完成所有上涨天数的统计
            if (info.getChangeRate() > 0) incrementCount.incrementAndGet();
        });
        log.info("最近5日相近的量：code:{} 1:{},2:{},3:{},4:{},5:{},6:{},7:{}  总上涨天数：{}", baseCode, pre1ConformCount, pre2ConformCount, pre3ConformCount, pre4ConformCount, pre5ConformCount, pre6ConformCount, pre7ConformCount, incrementCount.get());

        log.info("添加redismap值,key:{},code:{}", redisMapKey, baseCode);
        HorizontalInfo horizontalInfo = new HorizontalInfo(baseCode, queryDate, startDate, endDate, moreThanMa250Count.get(), pre1ConformCount.get(), pre2ConformCount.get(),
                pre3ConformCount.get(), pre4ConformCount.get(), pre5ConformCount.get(), pre6ConformCount.get(), pre7ConformCount.get(), incrementCount.get(), baseInfo.getConcept_name(), baseInfo.getCzxbj_pm(),
                baseInfo.getCzxbj_jbmgsyzzlfh(), baseInfo.getGzbj_pm(), baseInfo.getGzbj_peg(), baseInfo.getDbfxbj_pm(), baseInfo.getDbfxbj_roepj(), baseInfo.getGsgmzsz_pm(), baseInfo.getGsgmzsz_zsz());
        redisUtil.hPut(redisMapKey, baseCode, JSON.toJSONString(horizontalInfo));
        return true;
    }

    public List<HorizontalInfo> queryHorizontal(Boolean forceJob, String queryDate, String startDate, String endDate, String code, EsQueryPageReq queryPage, int startStepDays, int startStepFullLimitCount, int startStepRange, int calculateAvgCountN, int avgNDownRange, int avgNDownRMoreExcludeCount, int avgNUpRange, int avgNUpRMoreExcludeCount, int moreThanMa250Range, int near7daysRange, Boolean sortBy250, String conceptName, int topToNDownRange) {
        log.info("queryList queryDate={},startDate={},endDate={},range={},pageable.page={},pageable.size={}, sortName:{} ,sortOrder:{}", queryDate, startDate, endDate, near7daysRange, queryPage.getPageNum(),
                queryPage.getPageSize(), queryPage.getSortName(), queryPage.getSortOrder());
        Map<Object, Object> stockBasicMap = redisUtil.hGetAll("stock_basic");

        String redisMapKey = "horizontal_" + queryDate + "_" + startDate + "_" + endDate + "_" + near7daysRange;
        List<HorizontalInfo> infos = new ArrayList<>();
        Map<Object, Object> objectMap = redisUtil.hGetAll(redisMapKey);
        if (objectMap.size() == 0 || forceJob) {
            jobQueryHorizontalToRedis(queryDate, startDate, endDate, code, queryPage, startStepDays, startStepFullLimitCount, startStepRange, calculateAvgCountN, avgNDownRange, avgNDownRMoreExcludeCount, avgNUpRange, avgNUpRMoreExcludeCount, moreThanMa250Range, near7daysRange, topToNDownRange);
            return null;
        } else {
            objectMap.values().stream().forEach(info -> {
                HorizontalInfo horizontalInfo = JSON.parseObject((String) info, HorizontalInfo.class);
                StockBaseInfo stockBaseInfo = JSON.parseObject((String) stockBasicMap.get(horizontalInfo.getCode()), StockBaseInfo.class);
                if (!Objects.isNull(stockBaseInfo)) {
                    horizontalInfo.setName(stockBaseInfo.getName());
                }
                if (StringUtils.isNotBlank(conceptName)) {
                    if (!horizontalInfo.getConceptName().contains(conceptName)) return;
                }
                infos.add(horizontalInfo);
            });
            if (!sortBy250) {
                if (StringUtils.equals(queryPage.getSortName(), "moreThanMa250Count")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre1ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre1ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre2ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre2ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre3ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre3ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre4ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre4ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre5ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre5ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre6ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre6ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre7ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getPre7ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "incrementCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getIncrementCount).reversed());
                } else {
                    //infos.sort(Comparator.comparing(HorizontalInfo::getPre1ConformCount).reversed());
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre1ConformCount).reversed());
                }
            } else {
                if (StringUtils.equals(queryPage.getSortName(), "moreThanMa250Count")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getMoreThanMa250Count).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre1ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre1ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre2ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre2ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre3ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre3ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre4ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre4ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre5ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre5ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre6ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre6ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "pre7ConformCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre7ConformCount).reversed());
                } else if (StringUtils.equals(queryPage.getSortName(), "incrementCount")) {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getIncrementCount).reversed());
                } else {
                    infos.sort(Comparator.comparing(HorizontalInfo::getMoreThanMa250Count)
                            .thenComparing(HorizontalInfo::getPre1ConformCount).reversed());
                }
            }
            return infos;
        }
    }

}
